
package com.hexinfo.dmpro.dev.data.transfer.service.impl;

import cn.afterturn.easypoi.excel.ExcelExportUtil;
import cn.afterturn.easypoi.excel.entity.TemplateExportParams;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.base.PageResult;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.data.transfer.common.common.manager.AgentManager;
import com.hexinfo.dmpro.data.transfer.common.common.util.ShellUtil;
import com.hexinfo.dmpro.data.transfer.common.model.AgentInfo;
import com.hexinfo.dmpro.data.transfer.common.service.IAgentInfoService;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferImportTaskStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferListStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.dao.DevDataImportListMapper;
import com.hexinfo.dmpro.dev.data.transfer.handle.IDevDataImportHandler;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataExportList;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataImportList;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataImportTask;
import com.hexinfo.dmpro.dev.data.transfer.model.FtpServerModel;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataImportListService;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataImportTaskService;
import com.hexinfo.dmpro.dev.data.transfer.utils.ExcelUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.Workbook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 表导入任务表
 *
 * @author yyh.huang
 * @date 2023-10-25 19:15:04
 */
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class DevDataImportListServiceImpl extends ServiceImpl<DevDataImportListMapper, DevDataImportList> implements IDevDataImportListService {

    private final AgentManager agentManager;

    private final IDevDataImportTaskService devDataImportTaskService;

    private final IDevDataImportHandler devDataImportHandler;

    private final IAgentInfoService agentInfoService;

    @Value("${dev-data-ftp.server}")
    private String ftpServer;

    @Value("${dev-data-ftp.port}")
    private Integer ftpPort;

    @Value("${dev-data-ftp.user}")
    private String ftpUserName;

    @Value("${dev-data-ftp.password}")
    private String ftpPassword;

    @Value("${dev-data-import-retry-count}")
    private Integer retryCount;

    @Value("${script.pack.relativePath}")
    private String relativePath;

    @Override
    public PageResult queryList(Page page, DevDataImportList devDataImportList) {
        LambdaQueryWrapper<DevDataImportList> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(DevDataImportList::getDel, CommonConstants.STATUS_NORMAL);
        if (StrUtil.isNotBlank(devDataImportList.getTaskCode())) {
            wrapper.like(DevDataImportList::getTaskCode, devDataImportList.getTaskCode());
        }
        if (StrUtil.isNotBlank(devDataImportList.getDbName())) {
            wrapper.like(DevDataImportList::getDbName, devDataImportList.getDbName());
        }
        if (StrUtil.isNotBlank(devDataImportList.getTableName())) {
            wrapper.like(DevDataImportList::getTableName, devDataImportList.getTableName());
        }
        if (StrUtil.isNotBlank(devDataImportList.getErrorMessage())) {
            wrapper.like(DevDataImportList::getErrorMessage, devDataImportList.getErrorMessage());
        }
        if (StrUtil.isNotBlank(devDataImportList.getImportStatus())) {
            wrapper.eq(DevDataImportList::getImportStatus, devDataImportList.getImportStatus());
        }
        if (CollectionUtil.isNotEmpty(devDataImportList.getCreateTimeRange())) {
            Date beginTime = DateUtil.parse(devDataImportList.getCreateTimeRange().get(0));
            Date endTime = DateUtil.parse(devDataImportList.getCreateTimeRange().get(1));
            wrapper.between(DevDataImportList::getCreateTime, beginTime, endTime);
        }
        wrapper.orderByDesc(DevDataImportList::getCreateTime).orderByAsc(DevDataImportList::getDbName).orderByAsc(DevDataImportList::getTableName);
        Page returnPage = this.baseMapper.selectPage(page, wrapper);
        return new PageResult(returnPage.getRecords(), returnPage.getTotal());
    }

    @Override
    public List<DevDataImportList> queryListByTaskCode(String taskCode) {
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.eq("task_code", taskCode);
        return this.baseMapper.selectList(queryWrapper);
    }

    @Override
    @Transactional
    public void updateStatus(DevDataImportList devDataImportList) {
        DevDataImportList dbDevDataImportList = getById(devDataImportList.getId());
        if (DevDataTransferListStatusEnum.RUNNING.getValue().equals(dbDevDataImportList.getImportStatus())) {
            agentManager.removeExecuteTask(devDataImportList.getAgentKey());
            removeHdfsTmpDir(devDataImportList.getTaskCode());
            if (DevDataTransferListStatusEnum.FAIL.getValue().equals(devDataImportList.getImportStatus())
                    && dbDevDataImportList.getErrorCount() < retryCount) {
                exportRetryHandler(devDataImportList.getId());
            } else {
                exportFinishHandler(devDataImportList);
            }
        }
    }

    private void exportRetryHandler(String id) {
        this.baseMapper.updateByRetry(id);
        devDataImportHandler.execute(this.baseMapper.selectById(id));
    }

    private void exportFinishHandler(DevDataImportList devDataImportList) {
        // 为规避并发更新单条的数据问题，这边只做更新List状态，具体的数据汇总做到分页查询中
        updateDevDataImportListStatus(devDataImportList);
//        updateDevDataImportTaskStatus(devDataImportList);
    }

    private void updateDevDataImportListStatus(DevDataImportList devDataImportList) {
        devDataImportList.setEndTime(DateUtil.date());
        this.updateById(devDataImportList);
    }

    private void updateDevDataImportTaskStatus(DevDataImportList devDataImportList) {
        if (DevDataTransferListStatusEnum.SUCCESS.getValue().equals(devDataImportList.getImportStatus())) {
            devDataImportTaskService.modifySuccessNumByTaskCode(devDataImportList.getTaskCode());
        } else {
            devDataImportTaskService.modifyFailNumByTaskCode(devDataImportList.getTaskCode());
        }
        devDataImportTaskService.modifyStatusByTaskCode(devDataImportList.getTaskCode(), DateUtil.now());
    }

    private void removeHdfsTmpDir(String taskCode) {
        DevDataImportTask devDataImportTask = devDataImportTaskService.queryByTaskCode(taskCode);
        if (devDataImportTask != null && DevDataTransferImportTaskStatusEnum.finish.getValue().equals(devDataImportTask.getTaskStatus())) {
            String hdfsTmpRootPath = "/tmp/dmpro";
            removeHdfsTmpOutputDir(hdfsTmpRootPath, taskCode);
            removeHdfsTmpInputDir(hdfsTmpRootPath, taskCode);
        }
    }

    private void removeHdfsTmpOutputDir(String hdfsTmpRootPath, String taskCode) {
        String hdfsPath = StrUtil.format("{}/import/dev/output/{}", hdfsTmpRootPath, taskCode);
        executeRemoveHdfsTmpDir(hdfsPath);
    }

    private void removeHdfsTmpInputDir(String hdfsTmpRootPath, String taskCode) {
        String hdfsPath = StrUtil.format("{}/import/dev/input/{}", hdfsTmpRootPath, taskCode);
        executeRemoveHdfsTmpDir(hdfsPath);
    }

    private void executeRemoveHdfsTmpDir(String hdfsPath) {
        String[] shell = new String[]{"sudo", "-u", "hdfs", "hadoop", "fs", "-rm", "-r", "-f", hdfsPath};
        Map result = ShellUtil.exec(shell);
        if (!result.containsKey("INVOKE_STATUS") || !(boolean) result.get("INVOKE_STATUS")) {
            log.warn(StrUtil.format("删除HDFS目录发生异常: {}", result.get("SHELL_ERR_RESULT")));
        }
    }

    @Override
    public void restart(List<String> ids) {
        List<DevDataImportList> devDataExportLists = queryRestartListByIds(ids);
        for (DevDataImportList devDataImportList : devDataExportLists) {
            devDataImportHandler.execute(devDataImportList);
        }
    }

    @Override
    public void restart(String id) {
        DevDataImportList devDataImportList = this.baseMapper.selectById(id);
        devDataImportHandler.execute(devDataImportList);
    }

    @Override
    public int updateByRestart(String id) {
        return this.baseMapper.updateByRestart(id);
    }

    private List<DevDataImportList> queryRestartListByIds(List<String> ids) {
        List<DevDataImportList> devDataImportLists = new ArrayList<>();
        for (String id : ids) {
            DevDataImportList devDataImportList = this.getById(id);
            this.updateByRestart(id);
            if (DevDataTransferListStatusEnum.SUCCESS.getValue().equals(devDataImportList.getImportStatus())) {
                devDataImportTaskService.reduceSuccessNumByTaskCode(devDataImportList.getTaskCode());
            } else if (DevDataTransferListStatusEnum.FAIL.getValue().equals(devDataImportList.getImportStatus())) {
                devDataImportTaskService.reduceFailNumByTaskCode(devDataImportList.getTaskCode());
            }
            devDataImportTaskService.modifyStatusByTaskCode(devDataImportList.getTaskCode(), DateUtil.now());
            devDataImportLists.add(this.getById(id));
        }
        return devDataImportLists;
    }

    @Async
    public void buildImportTaskList(DevDataImportTask devDataImportTask, List<String> filePathList) {
        try {
            // ① 保存导出任务清单
            createDevDataImportList(devDataImportTask, filePathList);
            // ② 更新导入任务状态为进行中
            devDataImportTaskService.updateExportTaskStatusById(devDataImportTask.getId());
            // ③ 执行导出任务流程
            toExecuteImportTask(devDataImportTask);
        } catch (Exception e) {
            log.error(StrUtil.format("构建开发数据导入任务[{}]失败: {}", devDataImportTask.getTaskCode(), e.getMessage()), e);
            devDataImportTaskService.updateExportTaskFailById(devDataImportTask, e.getMessage());
        }
    }
    /**
     * @Method download
     * @Description 开发数据导入表清单下载
     * @Author gxd
     * @return String
     * @Date 2024/1/4 11:57
     **/
    @Override
    public String download(DevDataImportList devDataImportList) {
        String savePath = "";
        try {
            List<DevDataImportList> queryLists = query(devDataImportList);
            String exportPath = relativePath + "/import/";
            savePath = downTabliListFile(exportPath, queryLists);
        }catch (Exception e){
            e.printStackTrace();
        }
        return savePath;
    }

    /**
     * @Method abolishByTaskCode
     * @Param
     * @param taskCode
     * @Return void
     * @Description 根据TaskCode批量修改状态为中止
     * @Author Wang zhihao
     * @Date 2024/1/19 11:32
     * @Version V1.0
     */
    @Override
    public void abolishByTaskCode(String taskCode) {
        LambdaUpdateWrapper<DevDataImportList> wrapper = Wrappers.lambdaUpdate();
        if (StrUtil.isNotBlank(taskCode)) {
            wrapper.in(DevDataImportList::getTaskCode, taskCode);
            wrapper.in(DevDataImportList::getImportStatus, DevDataTransferListStatusEnum.TODO.getValue(), DevDataTransferListStatusEnum.RUNNING.getValue());
        }
        DevDataImportList devDataImportList = new DevDataImportList();
        devDataImportList.setImportStatus(DevDataTransferListStatusEnum.ABOLISH.getValue());
        update(devDataImportList, wrapper);
    }

    /**
     * @Method
     * @Description 生成本地excel文件
     * @Author gxd
     * @param[1] exportPath
     * @param[2] devDataImportLists
     * @return String
     * @Date 2024/1/5 9:19
     **/
    private String downTabliListFile(String exportPath, List<DevDataImportList> devDataImportLists) {
        Workbook workbook = null;
        File saveFile = null;
        String savePath = "";
        try {
            // 模板位置
            String templateFilePath = "static/import_result_template.xls";
            // 读取模板
            final TemplateExportParams templateExportParams = new TemplateExportParams(templateFilePath);
            // 获取导出数据
            Map<String, Object> sheetMap = new HashMap<>();
            sheetMap.put("devDataImportLists", devDataImportLists.stream().map(t -> JSON.parseObject(JSON.toJSONString(t), Map.class)).collect(Collectors.toList()));

            // 生成数据
            workbook = ExcelExportUtil.exportExcel(templateExportParams, sheetMap);
            // 创建文件
            savePath = exportPath + DateUtil.format(new DateTime(), "yyyyMMddHHmmss") + "_开发数据导入表清单.xls";
            saveFile = ExcelUtils.createDatafile(exportPath, savePath);
            // 写入数据
            ExcelUtils.writeDownloadFileData(saveFile, workbook);
        }catch (Exception e) {
            log.error("导出清单到服务器异常" + e.getMessage(), e);
            throw new RuntimeException("导出清单到服务器异常" + e.getMessage(), e);
        } finally {
            try {
                workbook.close();
            } catch (Exception e) {
                log.error("导出清单时关闭异常");
                e.printStackTrace();
            }
        }
        return savePath;
    }
    public List<DevDataImportList> query(DevDataImportList devDataImportList) {
        LambdaQueryWrapper<DevDataImportList> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(DevDataImportList::getDel, CommonConstants.STATUS_NORMAL);
        if (StrUtil.isNotBlank(devDataImportList.getTaskCode())) {
            wrapper.eq(DevDataImportList::getTaskCode, devDataImportList.getTaskCode());
        }
        if (StrUtil.isNotBlank(devDataImportList.getDbName())) {
            wrapper.like(DevDataImportList::getDbName, devDataImportList.getDbName());
        }
        if (StrUtil.isNotBlank(devDataImportList.getTableName())) {
            wrapper.like(DevDataImportList::getTableName, devDataImportList.getTableName());
        }
        if (StrUtil.isNotBlank(devDataImportList.getErrorMessage())) {
            wrapper.like(DevDataImportList::getErrorMessage, devDataImportList.getErrorMessage());
        }
        if (StrUtil.isNotBlank(devDataImportList.getImportStatus())) {
            wrapper.eq(DevDataImportList::getImportStatus, devDataImportList.getImportStatus());
        }
        wrapper.orderByAsc(DevDataImportList::getDbName).orderByAsc(DevDataImportList::getTableName);
        return this.baseMapper.selectList(wrapper);
    }
    private void createDevDataImportList(DevDataImportTask devDataImportTask, List<String> filePathList) {
        log.info("构建导入任务清单[{}]开始，开始时间:{}", devDataImportTask.getTaskCode(), DateUtil.now());
        for (String filePath : filePathList) {
            this.save(buildDevDataImportList(devDataImportTask, filePath));
        }
        log.info("构建导入任务清单[{}]开始，开始时间:{}", devDataImportTask.getTaskCode(), DateUtil.now());
    }

    private DevDataImportList buildDevDataImportList(DevDataImportTask devDataImportTask, String filePath) {
        String fileName = filePath.substring(filePath.lastIndexOf("/") + 1);
        String remoteDir = filePath.substring(0, filePath.lastIndexOf("/"));
        String[] fileNameArr = StrUtil.split(fileName, ".");
        String dbName = fileNameArr[0];
        String tableName = fileNameArr[1];
        DevDataImportList devDataImportList = new DevDataImportList();
        devDataImportList.setTaskCode(devDataImportTask.getTaskCode());
        devDataImportList.setDbName(dbName);
        devDataImportList.setTableName(tableName);
        devDataImportList.setFilePath(remoteDir);
        devDataImportList.setFileName(fileName);
        devDataImportList.setErrorCount(0);
        devDataImportList.setImportStatus(DevDataTransferListStatusEnum.TODO.getValue());
        devDataImportList.setSqlContent(getSqlContent(remoteDir));
        return devDataImportList;
    }

    private String getSqlContent(String remoteDir) {
        String content = readFile(remoteDir, "ddl.txt");
        String[] sqlList = StrUtil.split(content, ";");
        // 构建带SQL标识的sqlContent
        StringBuilder ddlSql = new StringBuilder();
        ddlSql.append("--创建Hive临时文本表SQL\n").append(sqlList.length > 0 ? sqlList[0] : "").append(";\n");
        ddlSql.append("--创建目标表SQL\n").append(sqlList.length > 1 ? sqlList[1] : "").append(";\n");
        ddlSql.append("--文本表数据插入到目标表SQL\n").append(sqlList.length > 2 ? sqlList[2] : "").append(";");
        return ddlSql.toString();
    }

    private String readFile(String filePath, String fileName) {
        try {
            AgentInfo agentInfo = getAvailableAgent();
            final HttpRequest post = HttpUtil.createPost(agentInfo.getFtpReadFileHttpUrl());
            FtpServerModel ftpServerModel = buildFtpServerModel(filePath);
            ftpServerModel.setFtpFileName(fileName);
            String body = JSON.toJSONString(ftpServerModel);
            log.info("读取FTP文件【{}.{}】内容请求参数: {}", filePath, fileName, body);
            final HttpResponse execute = post.body(body).execute();
            if (200 != execute.getStatus()) {
                throw new RuntimeException("HTTP请求失败: " + execute.body());
            }
            log.info("读取FTP文件【{}.{}】内容请求成功！(●'◡'●)", filePath, fileName);
            return execute.body();
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("获取指定FTP目录【{}】下所有文件异常: {}", filePath, e.getMessage()), e);
        }
    }

    private FtpServerModel buildFtpServerModel(String filePath) {
        return new FtpServerModel(ftpServer, ftpPort, ftpUserName, ftpPassword, filePath);
    }

    private AgentInfo getAvailableAgent() {
        List<AgentInfo> agentInfoList = agentInfoService.queryList(new AgentInfo());
        if (CollectionUtil.isEmpty(agentInfoList)) {
            throw new RuntimeException("没有可用的AGENT");
        }
        return agentInfoList.get(0);
    }

    private void toExecuteImportTask(DevDataImportTask devDataImportTask) {
        List<DevDataImportList> devDataImportLists = queryListByTaskCode(devDataImportTask.getTaskCode());
        for (DevDataImportList devDataImportList : devDataImportLists) {
            devDataImportHandler.execute(devDataImportList);
        }
    }

}
